001    /*
002     * Copyright 2005 Stephen J. McConnell.
003     *
004     * Licensed  under the  Apache License,  Version 2.0  (the "License");
005     * you may not use  this file  except in  compliance with the License.
006     * You may obtain a copy of the License at
007     *
008     *   http://www.apache.org/licenses/LICENSE-2.0
009     *
010     * Unless required by applicable law or agreed to in writing, software
011     * distributed  under the  License is distributed on an "AS IS" BASIS,
012     * WITHOUT  WARRANTIES OR CONDITIONS  OF ANY KIND, either  express  or
013     * implied.
014     *
015     * See the License for the specific language governing permissions and
016     * limitations under the License.
017     */
018    
019    package net.dpml.station.server; 
020    
021    import java.rmi.AlreadyBoundException;
022    import java.rmi.RemoteException;
023    import java.rmi.server.UnicastRemoteObject;
024    import java.rmi.registry.LocateRegistry;
025    import java.rmi.registry.Registry;
026    import java.net.URL;
027    import java.util.Map;
028    import java.util.Hashtable;
029    import java.util.LinkedList;
030    import java.util.List;
031    import java.util.EventObject;
032    
033    import net.dpml.station.Application;
034    import net.dpml.station.Callback;
035    import net.dpml.station.Manager;
036    import net.dpml.station.Station;
037    import net.dpml.station.StationException;
038    
039    import net.dpml.station.info.ApplicationDescriptor;
040    import net.dpml.station.info.StartupPolicy;
041    import net.dpml.station.ApplicationRegistry;
042    
043    import net.dpml.util.Logger;
044    import net.dpml.transit.model.TransitModel;
045    import net.dpml.transit.Disposable;
046    
047    import net.dpml.lang.UnknownKeyException;
048    
049    /**
050     * The RemoteStation is responsible for the establishment of 
051     * callback monitors to external processes established by the 
052     * station manager.
053     * @author <a href="http://www.dpml.net">Digital Product Meta Library</a>
054     * @version 1.0.1
055     */
056    public class RemoteStation extends UnicastRemoteObject implements Station, Manager
057    {
058        private final RemoteApplicationRegistry m_registry;
059        private final Map m_applications = new Hashtable();
060        private final Logger m_logger;
061        private final int m_port;
062        private final Registry m_rmiRegistry;
063        private final URL m_store;
064        private final TransitModel m_model;
065        private final LoggingServer m_server;
066        private final Thread m_thread;
067        
068        private boolean m_terminated = false;
069        
070       /**
071        * Creation of a station instance.
072        *
073        * @param logger the assigned logging channel
074        * @param model the transit model
075        * @param port the station port 
076        * @param registryStorageUrl uri defining the registry backing store
077        * @exception Exception if a exception occurs during establishment
078        */
079        public RemoteStation( 
080          Logger logger, TransitModel model, int port, URL registryStorageUrl ) 
081          throws Exception
082        {
083            super();
084            
085            m_logger = logger;
086            m_port = port;
087            m_store = registryStorageUrl;
088            m_model = model;
089            
090            m_rmiRegistry = getLocalRegistry( port );
091            
092            try
093            {
094                m_rmiRegistry.bind( STATION_KEY, this );
095            }
096            catch( AlreadyBoundException e )
097            {
098                final String error =
099                 "An instance of the Station is already bound to port " + port;
100                throw new StationException( error, e );
101            }
102    
103            setShutdownHook( this );
104            startEventDispatchThread();
105    
106            try
107            {
108                m_server = new LoggingServer( 2020 );
109                m_thread = new Thread( m_server, "DPML Station Logging Server" );
110                m_thread.start();
111            }
112            catch( Exception e )
113            {
114                final String error =
115                 "Unexpected error while attempting to start the logging server on port " + 2020;
116                throw new StationException( error, e );
117            }
118            
119            if( getLogger().isDebugEnabled() )
120            {
121                if( null == registryStorageUrl )
122                {
123                    getLogger().debug( "loading registry from default storage" );
124                }
125                else
126                {
127                    getLogger().debug( "loading registry from [" + registryStorageUrl + "]" );
128                }
129            }
130            
131            m_registry = new RemoteApplicationRegistry( logger, registryStorageUrl );
132            String[] keys = m_registry.getKeys();
133            
134            if( getLogger().isDebugEnabled() )
135            {
136                getLogger().debug( "registry established (" + keys.length + ")" );
137            }
138            
139            for( int i=0; i<keys.length; i++ )
140            {
141                String key = keys[i];
142                try
143                {
144                    ApplicationDescriptor descriptor = 
145                      m_registry.getApplicationDescriptor( key );
146                    if( StartupPolicy.AUTOMATIC.equals( descriptor.getStartupPolicy() ) )
147                    {
148                        RemoteApplication application = getRemoteApplication( key );
149                        application.start();
150                    }
151                }
152                catch( UnknownKeyException e )
153                {
154                    throw new RuntimeException( e ); // will not happen
155                }
156            }
157        }
158        
159       /**
160        * Return a string containing info about the general setup of the station.
161        * @return station configuration info
162        */
163        public String[] getInfo()
164        {
165            String[] values = new String[4];
166            values[0] = "Port: " + m_port;
167            values[1] = "Store: " + m_store;
168            values[2] = "Basedir: " + System.getProperty( "user.dir" );
169            values[3] = "Codebase: " 
170              + getClass().getProtectionDomain().getCodeSource().getLocation();
171            return values;
172        }
173        
174       /**
175        * Return an callback handler for the supplied id.
176        * @param id the callback id
177        * @return the callback handler
178        * @exception UnknownKeyException if the id is unknown
179        * @exception RemoteException if a remote error occurs
180        */
181        public Callback getCallback( String id ) throws UnknownKeyException, RemoteException
182        {
183            // TODO: improve this so that this is only called once per appliation
184            return getRemoteApplication( id );
185        }
186        
187       /**
188        * Shutdown the station.
189        */
190        public void shutdown()
191        {
192            shutdown( true );
193        }
194        
195       /**
196        * Shutdown the station.
197        * @param exit if true launch a process termination
198        */
199        private void shutdown( boolean exit )
200        {
201            synchronized( m_applications )
202            {
203                if( m_terminated )
204                {
205                    return;
206                }
207                else
208                {
209                    m_terminated = true;
210                }
211                
212                if( getLogger().isInfoEnabled() )
213                {
214                    getLogger().info( "initiating station shutdown" );
215                }
216                
217                try
218                {
219                    m_rmiRegistry.unbind( STATION_KEY );
220                }
221                catch( Exception e )
222                {
223                    // ignore
224                }
225                try
226                {
227                    RemoteApplication[] applications = getRemoteApplications();
228                    for( int i=0; i<applications.length; i++ )
229                    {
230                        RemoteApplication application = applications[i];
231                        application.shutdown();
232                        UnicastRemoteObject.unexportObject( application, true );
233                    }
234                    UnicastRemoteObject.unexportObject( m_registry, true );
235                }
236                catch( Exception e )
237                {
238                    // ignore
239                }
240                try
241                {
242                    int n =  m_server.getErrorCount();
243                    if( n > 0 )
244                    {
245                        getLogger().warn( "logging issues: " + n );
246                    }
247                    m_thread.interrupt();
248                }
249                catch( Exception e )
250                {
251                    // ignore
252                }
253                
254                finally
255                {
256                    if( getLogger().isInfoEnabled() )
257                    {
258                        getLogger().info( "station shutdown complete" );
259                    }
260                    
261                    if( exit )
262                    {
263                        if( m_model instanceof Disposable )
264                        {
265                            try
266                            {
267                                Disposable disposable = (Disposable) m_model;
268                                disposable.dispose();
269                            }
270                            catch( Exception e )
271                            {
272                                // ignore
273                            }
274                        }
275                        
276                        if( getLogger().isDebugEnabled() )
277                        {
278                            getLogger().debug( "terminating process" );
279                        }
280                        
281                        Thread thread = new Thread(
282                          new Runnable()
283                          {
284                            public void run()
285                            {
286                                RemoteStation.m_DISPATCH.dispose();
287                                System.exit( 0 );
288                            }
289                          }
290                        );
291                        thread.start();
292                    }
293                }
294            }
295        }
296        
297       /**
298        * Return the application registry.
299        * @return the registry
300        */
301        public ApplicationRegistry getApplicationRegistry()
302        {
303            return m_registry;
304        }
305    
306       /**
307        * Return an application reference for the supplied key.
308        * @param key the application key
309        * @return the application
310        * @exception UnknownKeyException if the key is unknown
311        * @exception RemoteException if a remote error occurs
312        */
313        public Application getApplication( String key ) throws UnknownKeyException, RemoteException
314        {
315            return getRemoteApplication( key );
316        }
317        
318       /**
319        * Return an application reference for the supplied key.
320        * @param key the application key
321        * @return the application
322        * @exception UnknownKeyException if the key is unknown
323        * @exception RemoteException if a remote error occurs
324        */
325        RemoteApplication getRemoteApplication( String key ) throws UnknownKeyException, RemoteException
326        {
327            synchronized( m_applications )
328            {
329                if( m_applications.containsKey( key ) )
330                {
331                    return (RemoteApplication) m_applications.get( key );
332                }
333                else
334                {
335                    Logger logger = getLogger().getChildLogger( key );
336                    ApplicationDescriptor descriptor = m_registry.getApplicationDescriptor( key );
337                    RemoteApplication application = 
338                      new RemoteApplication( logger, descriptor, key, m_port );
339                    m_applications.put( key, application );
340                    return application;
341                }
342            }
343        }
344    
345       /**
346        * Return an array of all remote applications.
347        * @return the applications array
348        */
349        RemoteApplication[] getRemoteApplications()
350        {
351            synchronized( m_applications )
352            {
353                return (RemoteApplication[]) m_applications.values().toArray( new RemoteApplication[0] );
354            }
355        }
356        
357        private Logger getLogger()
358        {
359            return m_logger;
360        }
361        
362        private Registry getLocalRegistry( int port ) throws RemoteException
363        {
364            try
365            {
366                Registry registry = LocateRegistry.createRegistry( port );
367                getLogger().debug( "created local registry on port " + port );
368                return registry;
369            }
370            catch( RemoteException e )
371            {
372                Registry registry = LocateRegistry.getRegistry( port );
373                getLogger().debug( "using local registry on port " + port );
374                return registry;
375            }
376        }
377    
378        /**
379         * Queue of pending notification events.  When an event for which 
380         * there are one or more listeners occurs, it is placed on this queue 
381         * and the queue is notified.  A background thread waits on this queue 
382         * and delivers the events.  This decouples event delivery from 
383         * the application concern, greatly simplifying locking and reducing 
384         * opportunity for deadlock.
385         */
386        private static final List EVENT_QUEUE = new LinkedList();
387    
388       /**
389        * Enqueue an event for delivery to registered
390        * listeners unless there are no registered
391        * listeners.
392        * @param event the event to enqueue
393        */
394        static void enqueueEvent( EventObject event )
395        {
396            synchronized( EVENT_QUEUE ) 
397            {
398                EVENT_QUEUE.add( event );
399                EVENT_QUEUE.notify();
400            }
401        }
402        
403        /**
404         * A single background thread ("the event notification thread") monitors
405         * the event queue and delivers events that are placed on the queue.
406         */
407        private static class EventDispatchThread extends Thread 
408        {
409            private final Logger m_logger;
410            
411            private boolean m_continue = true;
412            
413            EventDispatchThread( Logger logger )
414            {
415                super( "DPML Station Event Dispatch" );
416                m_logger = logger;
417                m_logger.debug( "starting event dispatch thread" );
418            }
419            
420            void dispose()
421            {
422                synchronized( EVENT_QUEUE )
423                {
424                    m_logger.debug( "stopping event dispatch thread" );
425                    m_continue = false;
426                    EVENT_QUEUE.notify();
427                }
428            }
429            
430            public void run() 
431            {
432                while( m_continue ) 
433                {
434                    // Wait on EVENT_QUEUE till an event is present
435                    EventObject event = null;
436                    synchronized( EVENT_QUEUE ) 
437                    {
438                        try
439                        {
440                            while( EVENT_QUEUE.isEmpty() )
441                            {
442                                EVENT_QUEUE.wait();
443                            }
444                            Object object = EVENT_QUEUE.remove( 0 );
445                            try
446                            {
447                                event = (EventObject) object;
448                            }
449                            catch( ClassCastException cce )
450                            {
451                                final String error = 
452                                  "Unexpected class cast exception while processing an event." 
453                                  + "\nEvent: " + object;
454                                throw new IllegalStateException( error );
455                            }
456                        }
457                        catch( InterruptedException e )
458                        {
459                            return;
460                        }
461                    }
462                    
463                    Object source = event.getSource();
464                    if( source instanceof UnicastEventSource )
465                    {
466                        UnicastEventSource producer = (UnicastEventSource) source;
467                        try
468                        {
469                            producer.processEvent( event );
470                        }
471                        catch( Throwable e )
472                        {
473                            final String error = 
474                              "Unexpected error while processing event."
475                              + "\nEvent: " + event
476                              + "\nSource: " + source;
477                            m_logger.warn( error, e );
478                        }
479                    }
480                    else
481                    {
482                        final String error = 
483                          "Event source [" 
484                          + source.getClass().getName()
485                          + "] is not an instance of " + UnicastEventSource.class.getName();
486                        throw new IllegalStateException( error );
487                    }
488                }
489                
490                m_logger.info( "Controller event queue terminating." );
491            }
492        }
493    
494        private static EventDispatchThread m_DISPATCH = null;
495    
496        /**
497         * This method starts the event dispatch thread the first time it
498         * is called.  The event dispatch thread will be started only
499         * if someone registers a listener.
500         */
501        private synchronized void startEventDispatchThread()
502        {
503            if( m_DISPATCH == null )
504            {
505                Logger logger = getLogger();
506                m_DISPATCH = new EventDispatchThread( logger );
507                m_DISPATCH.setDaemon( true );
508                m_DISPATCH.start();
509            }
510        }
511            
512       /**
513        * Create a shutdown hook that will trigger shutdown of the supplied plugin.
514        * @param station the station
515        */
516        public static void setShutdownHook( final RemoteStation station )
517        {
518            //
519            // Create a shutdown hook to trigger clean disposal of the
520            // controller
521            //
522            
523            Runtime.getRuntime().addShutdownHook(
524              new Thread()
525              {
526                  public void run()
527                  {
528                      try
529                      {
530                          station.shutdown();
531                      }
532                      catch( Throwable e )
533                      {
534                          System.err.println( e.toString() );
535                      }
536                      System.runFinalization();
537                  }
538              }
539            );
540        }
541        
542    }